/*
 * Copyright (c) 2017 Baidu, Inc. All Rights Reserve.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package cn.trigram.uuid.bd.impl;

import cn.trigram.asserts.AssertUtil;
import cn.trigram.uuid.bd.BitsAllocator;
import cn.trigram.uuid.bd.UidGenerator;
import cn.trigram.uuid.bd.buffer.BufferPaddingExecutor;
import cn.trigram.uuid.bd.buffer.RejectedPutBufferHandler;
import cn.trigram.uuid.bd.buffer.RejectedTakeBufferHandler;
import cn.trigram.uuid.bd.buffer.RingBuffer;
import cn.trigram.uuid.bd.exception.UidGenerateException;
import cn.trigram.uuid.bd.worker.WorkerIdAssigner;
import java.util.ArrayList;
import java.util.List;
import javax.annotation.PreDestroy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Represents a cached implementation of {@link UidGenerator} extends
 * from {@link DefaultUidGenerator}, based on a lock free {@link RingBuffer}<p>
 *
 * The spring properties you can specified as below:<br>
 * <li><b>boostPower:</b> RingBuffer size boost for a power of 2, Sample: boostPower is 3, it means
 * the buffer size
 * will be <code>({@link BitsAllocator#getMaxSequence()} + 1) &lt;&lt;
 * {@link #boostPower}</code>, Default as {@value #DEFAULT_BOOST_POWER}
 * <li><b>paddingFactor:</b> Represents a percent value of (0 - 100). When the count of rest
 * available UIDs reach the
 * threshold, it will trigger padding buffer. Default as{@link RingBuffer#DEFAULT_PADDING_PERCENT}
 * Sample: paddingFactor=20, bufferSize=1000 -> threshold=1000 * 20 /100, padding buffer will be
 * triggered when tail-cursor<threshold
 * <li><b>scheduleInterval:</b> Padding buffer in a schedule, specify padding buffer interval, Unit
 * as second
 * <li><b>rejectedPutBufferHandler:</b> Policy for rejected put buffer. Default as discard put
 * request, just do logging
 * <li><b>rejectedTakeBufferHandler:</b> Policy for rejected take buffer. Default as throwing up an
 * exception
 *
 * @author yutianbao
 */
public class CachedUidGenerator extends DefaultUidGenerator {

  private static final Logger LOGGER = LoggerFactory.getLogger(CachedUidGenerator.class);

  private static final int DEFAULT_BOOST_POWER = 3;

  /* Spring properties */

  /**
   * 对于RingBuffer缓存的容量，为最大序号容量的2^n倍，默认为3。 通过左移 n 位
   */
  private int boostPower = DEFAULT_BOOST_POWER;

  /**
   * 指定何时向RingBuffer中填充UID, 取值为百分比(0, 100), 默认为50
   * <br/>
   * 举例: bufferSize=1024, paddingFactor=50 -> threshold=1024 * 50 / 100 = 512.
   * 当环上可用UID数量 < 512时, 将自动对RingBuffer进行填充补全
   */
  private int paddingFactor = RingBuffer.DEFAULT_PADDING_PERCENT;

  /**
   * 另外一种RingBuffer填充时机, 在Schedule线程中, 周期性检查填充
   * <br/>
   * 默认:不配置此项, 即不实用Schedule线程. 如需使用, 请指定Schedule线程时间间隔, 单位:秒
   */
  private Long scheduleInterval;

  /**
   * 拒绝策略: 当环已满, 无法继续填充时
   * <br/>
   * 默认无需指定, 将丢弃Put操作, 仅日志记录. 如有特殊需求, 请实现RejectedPutBufferHandler接口(支持Lambda表达式)
   */
  private RejectedPutBufferHandler rejectedPutBufferHandler;

  /**
   * 拒绝策略: 当环已空, 无法继续获取时
   * <br/>
   * 默认无需指定, 将记录日志, 并抛出UidGenerateException异常. 如有特殊需求, 请实现RejectedTakeBufferHandler接口(支持Lambda表达式)
   */
  private RejectedTakeBufferHandler rejectedTakeBufferHandler;

  /** RingBuffer */
  private RingBuffer ringBuffer;

  private BufferPaddingExecutor bufferPaddingExecutor;

  public CachedUidGenerator(int timeBits, int workerBits, int seqBits,
      WorkerIdAssigner workerIdAssigner, Long scheduleInterval,
      RejectedPutBufferHandler rejectedPutBufferHandler,
      RejectedTakeBufferHandler rejectedTakeBufferHandler) {
    // initialize workerId & bitsAllocator
    super(timeBits, workerBits, seqBits, workerIdAssigner);
    this.scheduleInterval = scheduleInterval;
    this.rejectedPutBufferHandler = rejectedPutBufferHandler;
    this.rejectedTakeBufferHandler = rejectedTakeBufferHandler;
    // initialize RingBuffer & RingBufferPaddingExecutor
    this.initRingBuffer();
    LOGGER.info("Initialized RingBuffer successfully.");
  }

  public CachedUidGenerator(int timeBits, int workerBits, int seqBits,
      WorkerIdAssigner workerIdAssigner, Long scheduleInterval) {

    this(timeBits, workerBits, seqBits, workerIdAssigner, scheduleInterval, null, null);
  }

  @Override
  public long getUID() {

    try {
      return ringBuffer.take();
    } catch (Exception e) {
      LOGGER.error("Generate unique id exception. ", e);
      throw new UidGenerateException(e);
    }
  }

  @Override
  public String parseUID(long uid) {

    return super.parseUID(uid);
  }

  @PreDestroy
  public void destroy() {

    bufferPaddingExecutor.shutdown();
  }

  @Override
  protected void finalize() {

    this.destroy();
  }

  /**
   * 获取在指定秒内的最大序号容量数量的UID
   *
   * @return UID list, size of {@link BitsAllocator#getMaxSequence()} + 1
   */
  protected List<Long> nextIdsForOneSecond(long currentSecond) {
    // Initialize result list size of (max sequence + 1)
    int listSize = (int) bitsAllocator.getMaxSequence() + 1;
    List<Long> uidList = new ArrayList<>(listSize);

    // Allocate the first sequence of the second, the others can be calculated with the offset
    long firstSeqUid = bitsAllocator.allocate(currentSecond - epochSeconds, workerId, 0L);
    for (int offset = 0; offset < listSize; offset++) {
      uidList.add(firstSeqUid + offset);
    }

    return uidList;
  }

  /**
   * Initialize RingBuffer & RingBufferPaddingExecutor
   */
  private void initRingBuffer() {
    // initialize RingBuffer
    int bufferSize = ((int) bitsAllocator.getMaxSequence() + 1) << boostPower;
    this.ringBuffer = new RingBuffer(bufferSize, paddingFactor);
    LOGGER.info("Initialized ring buffer size:{}, paddingFactor:{}", bufferSize, paddingFactor);

    // initialize RingBufferPaddingExecutor
    boolean usingSchedule = (scheduleInterval != null);
    this.bufferPaddingExecutor = new BufferPaddingExecutor(ringBuffer, this::nextIdsForOneSecond,
        usingSchedule);
    if (usingSchedule) {
      bufferPaddingExecutor.setScheduleInterval(scheduleInterval);
    }

    LOGGER.info("Initialized BufferPaddingExecutor. Using schdule:{}, interval:{}", usingSchedule,
        scheduleInterval);

    // set rejected put/take handle policy
    this.ringBuffer.setBufferPaddingExecutor(bufferPaddingExecutor);
    if (rejectedPutBufferHandler != null) {
      this.ringBuffer.setRejectedPutHandler(rejectedPutBufferHandler);
    }
    if (rejectedTakeBufferHandler != null) {
      this.ringBuffer.setRejectedTakeHandler(rejectedTakeBufferHandler);
    }

    // fill in all slots of the RingBuffer
    bufferPaddingExecutor.paddingBuffer();

    // start buffer padding threads
    bufferPaddingExecutor.start();
  }

  /**
   * Setters for spring property
   */
  public void setBoostPower(int boostPower) {

    AssertUtil.assertTrue(boostPower > 0, "Boost power must be positive!");
    this.boostPower = boostPower;
  }

  public void setRejectedPutBufferHandler(RejectedPutBufferHandler rejectedPutBufferHandler) {

    AssertUtil.assertNotNull(rejectedPutBufferHandler, "RejectedPutBufferHandler can't be null!");
    this.rejectedPutBufferHandler = rejectedPutBufferHandler;
  }

  public void setRejectedTakeBufferHandler(RejectedTakeBufferHandler rejectedTakeBufferHandler) {

    AssertUtil.assertNotNull(rejectedTakeBufferHandler, "RejectedTakeBufferHandler can't be null!");
    this.rejectedTakeBufferHandler = rejectedTakeBufferHandler;
  }

  public void setScheduleInterval(long scheduleInterval) {

    AssertUtil.assertTrue(scheduleInterval > 0, "Schedule interval must positive!");
    this.scheduleInterval = scheduleInterval;
  }

}
